import random
import string
import sys
import time

from PyQt5 import QtWidgets
from PyQt5.QtCore import QThread, pyqtSignal, Qt, QTimer
from PyQt5.QtWidgets import QMainWindow, QVBoxLayout, QApplication
from kafka import KafkaConsumer
from kafka.errors import KafkaError

from KAFKALogs import global_log
from utils.utils import QLinedit_format


class BaseWindow(QtWidgets.QDialog):
    def __init__(self, parent=None):
        super().__init__(parent)
        self.snap_timer = QTimer()
        self.snap_timer.timeout.connect(self.snap_to_edge)

    def start_snap_timer(self):
        self.snap_timer.start(10)

    def snap_to_edge(self):
        desktop = QApplication.desktop()
        screen_rect = desktop.screenGeometry(desktop.screenNumber(self))
        window_rect = self.frameGeometry()

        distance = 18
        snap_positions = []

        if abs(window_rect.left() - screen_rect.left()) < distance:
            snap_positions.append((screen_rect.left(), window_rect.top()))
        if abs(window_rect.right() - screen_rect.right()) < distance:
            snap_positions.append((screen_rect.right() - window_rect.width(), window_rect.top()))
        if abs(window_rect.top() - screen_rect.top()) < distance:
            snap_positions.append((window_rect.left(), screen_rect.top()))
        if abs(window_rect.bottom() - screen_rect.bottom()) < distance:
            snap_positions.append((window_rect.left(), screen_rect.bottom() - window_rect.height()))

        if snap_positions:
            best_position = min(snap_positions,
                                key=lambda pos: abs(pos[0] - window_rect.left()) + abs(pos[1] - window_rect.top()))
            self.move(best_position[0], best_position[1])


class QthreadConsumer(QThread):
    signal_msg = pyqtSignal(str)
    signal_error = pyqtSignal(str)
    signal_status = pyqtSignal(bool)

    def __init__(self, host, port, topic, queue):
        super().__init__()
        self.consumer = None
        self.host = host
        self.port = port
        self.topic = topic
        self.queue = queue

    def run(self):
        """
        连接到kafka
        @return:
        """
        self.connect_to_kafka()
        self.status()
        self.exec_()

    def connect_to_kafka(self):
        """
        连接到kafka
        @return:
        """
        try:
            # 连接指定topic
            if self.topic:
                self.consumer = KafkaConsumer(self.topic, bootstrap_servers="%s:%s" % (self.host, self.port))
            # 连接指定queue
            if self.queue:
                self.consumer = KafkaConsumer(bootstrap_servers="%s:%s" % (self.host, self.port))
                self.consumer.subscribe(self.queue)
        # 连接失败
        except KafkaError as e:
            self.signal_error.emit(str(e))
        else:
            self.signal_msg.emit("已连接到服务器")
            self.receive_msg()

    def close_kafka(self):
        """
        关闭kafka
        @return:
        """
        if self.consumer:
            self.consumer.close()
            self.signal_msg.emit("已断开服务器连接")

    def receive_msg(self):
        """
        接收消息
        @return:
        """
        self.signal_msg.emit("开始接收消息")
        try:
            for msg in self.consumer:
                self.signal_msg.emit(msg.value.decode())
            self.status()
        except KafkaError as e:
            self.signal_error.emit(str(e))

    def clear_received_messages(self):
        """
        清空接收到的消息
        @return:
        """
        self.signal_msg.emit("清空接收到的消息")
        self.consumer = None

    def status(self):
        """
        返回连接状态
        @return:
        """
        if self.consumer:
            self.signal_status.emit(True)
        else:
            self.signal_status.emit(False)

    def stop_thread(self):
        """
        关闭线程
        @return:
        """
        self.terminate()
        self.status()

    def reconnect(self):
        """
        重新连接
        @return:
        """
        self.close_kafka()
        self.connect_to_kafka()
        self.status()


class WindowProduce(BaseWindow):
    def __init__(self, window_num, parent):
        super().__init__()

        self.producer = None
        self.windows = parent
        self.windows_num = window_num

        self.setWindowTitle("KAFKA生产者模拟工具 %s" % self.windows_num)

        self.host = QtWidgets.QLineEdit()
        self.host.setPlaceholderText("KAFKA服务器地址")
        self.host.setInputMask("000.000.000.000;_")

        self.port = QtWidgets.QLineEdit()
        validator = QLinedit_format(is_letter=False)  # 创建整数验证器
        self.port.setValidator(validator)  # 仅允许输入数字
        self.port.setPlaceholderText("端口")

        self.topic = QtWidgets.QLineEdit()
        self.topic.setPlaceholderText("TOPIC")

        self.queue = QtWidgets.QLineEdit()
        self.queue.setPlaceholderText("QUEUE")

        self.connect_button = QtWidgets.QPushButton("连接", self)
        self.disconnect_button = QtWidgets.QPushButton("断开连接", self)
        self.disconnect_button.setEnabled(False)

        self.send_button = QtWidgets.QPushButton("发送", self)
        self.send_button.setEnabled(False)

        self.clear_button = QtWidgets.QPushButton("清空消息", self)

        self.receive_msg_edit = QtWidgets.QTextEdit()
        self.receive_msg_edit.setReadOnly(True)

        layout1 = QtWidgets.QHBoxLayout()
        layout1.addWidget(QtWidgets.QLabel("地址："))
        layout1.addWidget(self.host)
        layout1.addWidget(QtWidgets.QLabel("端口："))
        layout1.addWidget(self.port)
        layout1.addWidget(QtWidgets.QLabel("TOPIC："))
        layout1.addWidget(self.topic)
        layout1.addWidget(QtWidgets.QLabel("QUEUE："))
        layout1.addWidget(self.queue)
        layout1.addWidget(self.connect_button)
        layout1.addWidget(self.disconnect_button)
        layout1.addWidget(self.send_button)
        layout1.addWidget(self.clear_button)

        layout2 = QtWidgets.QVBoxLayout()
        layout2.addWidget(QtWidgets.QLabel("接收到的消息："))
        layout2.addWidget(self.receive_msg_edit)

        main_layout = QtWidgets.QVBoxLayout()
        main_layout.addLayout(layout1)
        main_layout.addLayout(layout2)
        self.setLayout(main_layout)

        self.ui_control()

    def ui_control(self):
        """
        控制界面按钮状态
        @return:
        """
        self.connect_button.clicked.connect(self.connect_to_kafka)
        self.disconnect_button.clicked.connect(self.disconnect_to_kafka)

    def connect_to_kafka(self):
        """
        连接到kafka
        @return:
        """
        if (not self.topic.text() or not self.queue.text()) or (not self.host.text() and not self.port.text()):
            QtWidgets.QMessageBox.warning(self, "警告", "请填写完整信息")
            return
        self.producer = QthreadConsumer(self.host.text(), self.port.text(), self.topic.text(), self.queue.text())
        self.producer.start()
        self.producer.signal_msg.connect(self.ui_receive_msg)
        self.producer.signal_error.connect(self.ui_error)
        self.producer.signal_status.connect(self.ui_status)

    def disconnect_to_kafka(self):
        """
        断开kafka连接
        @return:
        """
        self.producer.close_kafka()
        self.ui_disconnect_to_kafka()

    def ui_connect_to_kafka(self):
        """
        连接到kafka
        @return:
        """
        self.connect_button.setEnabled(False)
        self.disconnect_button.setEnabled(True)

    def ui_disconnect_to_kafka(self):
        """
        断开kafka连接
        @return:
        """
        self.connect_button.setEnabled(True)
        self.disconnect_button.setEnabled(False)

    def ui_status(self, status):
        """
        连接状态
        @return:
        """
        if status:
            # 弹窗询问是否重连
            __quit = QtWidgets.QMessageBox.question(self, "消息", "是否重连？",
                                                    QtWidgets.QMessageBox.Yes | QtWidgets.QMessageBox.No,
                                                    QtWidgets.QMessageBox.No)
            if __quit == QtWidgets.QMessageBox.Yes:
                global_log.debug(f"{self.windows_num}窗口，重连到KAFKA服务器：{self.host.text()}:{self.port.text()}")
                self.producer.reconnect()
            else:
                self.producer.stop_thread()
                self.ui_disconnect_to_kafka()
        else:
            self.ui_disconnect_to_kafka()

    def ui_error(self, msg):
        """
        错误信息
        @return:
        """
        QtWidgets.QMessageBox.warning(self, "错误", msg)

    def ui_receive_msg(self, msg):
        """
        接收消息
        @return:
        """
        self.receive_msg_edit.insertPlainText(msg + "\n")
        self.receive_msg_edit.moveCursor(msg.End)
        self.receive_msg_edit.ensureCursorVisible()

    # 关闭窗口
    def closeEvent(self, a0):
        self.windows.remove_producer_window(self.windows_num)
        a0.accept()


class WindowConsumer(BaseWindow):
    def __init__(self, num_consumer, parent=None):
        super().__init__(parent)
        self.snap_timer = QTimer()
        self.snap_timer.timeout.connect(self.snap_to_edge)

        self.windows = parent
        self.windows_num = num_consumer

        self.setWindowTitle("KAFKA消费者模拟工具 %s" % num_consumer)

        self.host = QtWidgets.QLineEdit(self)
        self.host.setPlaceholderText("KAFKA服务器地址")
        self.host.setInputMask("000.000.000.000;_")

        self.port = QtWidgets.QLineEdit(self)
        self.port.setPlaceholderText("19092")
        validator = QLinedit_format(is_letter=False)  # 创建整数验证器
        self.port.setValidator(validator)  # 仅允许输入数字

        self.topic = QtWidgets.QLineEdit(self)
        self.topic.setPlaceholderText("TOPIC")

        self.queue = QtWidgets.QLineEdit(self)
        self.queue.setPlaceholderText("QUEUE")

        self.connect_button = QtWidgets.QPushButton("连接", self)
        self.disconnect_button = QtWidgets.QPushButton("断开连接", self)
        self.disconnect_button.setEnabled(False)
        self.close_button = QtWidgets.QPushButton("清空消息", self)

        self.receive_msg_edit = QtWidgets.QTextEdit(self)
        self.receive_msg_edit.setReadOnly(True)

        layout1 = QtWidgets.QHBoxLayout()
        layout1.addWidget(QtWidgets.QLabel("地址："))
        layout1.addWidget(self.host)
        layout1.addWidget(QtWidgets.QLabel("端口："))
        layout1.addWidget(self.port)
        layout1.addWidget(QtWidgets.QLabel("TOPIC："))
        layout1.addWidget(self.topic)
        layout1.addWidget(QtWidgets.QLabel("QUEUE："))
        layout1.addWidget(self.queue)
        layout1.addWidget(self.connect_button)
        layout1.addWidget(self.disconnect_button)
        layout1.addWidget(self.close_button)

        layout2 = QtWidgets.QVBoxLayout()
        layout2.addWidget(QtWidgets.QLabel("接收到的消息："))
        layout2.addWidget(self.receive_msg_edit)

        main_layout = QtWidgets.QVBoxLayout()
        main_layout.addLayout(layout1)
        main_layout.addLayout(layout2)
        self.setLayout(main_layout)

        self.connect_button.clicked.connect(self.connect_to_kafka)
        self.close_button.clicked.connect(self.clear_received_messages)

    def connect_to_kafka(self):
        # 连接到Kafka的逻辑
        pass

    def clear_received_messages(self):
        self.receive_msg_edit.clear()

    def closeEvent(self, a0):
        self.windows.remove_consumer_window(self.windows_num)
        a0.accept()


class MainWindow(QMainWindow, BaseWindow):
    def __init__(self):
        super().__init__()

        self.producer_windows_obj: list = []
        self.consumer_windows_obj: list = []
        self.setWindowFlags(self.windowFlags() | Qt.WindowStaysOnTopHint)

        self.producer_windows = {}
        self.consumer_windows = {}
        self.num_producer = 0
        self.num_consumer = 0

        # producer窗口的坐标
        self.x_producer = 0
        self.y_producer = 0

        # consumer窗口的坐标
        self.x_consumer = 0
        self.y_consumer = 0

        self.setWindowTitle("主窗口")
        self.start_producer_button = QtWidgets.QPushButton("启动生产者")
        self.start_producer_button.clicked.connect(self.start_producer)

        self.start_consumer_button = QtWidgets.QPushButton("启动消费者")
        self.start_consumer_button.clicked.connect(self.start_consumer)

        layout = QVBoxLayout()
        layout.addWidget(self.start_producer_button)
        layout.addWidget(self.start_consumer_button)

        central_widget = QtWidgets.QWidget()
        central_widget.setLayout(layout)
        self.setCentralWidget(central_widget)

    def start_producer(self):
        """
        启动服务器
        @return:
        """
        while True:
            __producer_name = self.generate_random_name()  # 生成随机名称
            if __producer_name not in self.producer_windows:  # 如果名称不在生产者窗口中
                producer_window = WindowProduce(window_num=__producer_name, parent=self)
                self.producer_windows[__producer_name] = producer_window
                producer_window.show()
                global_log.debug(f"生产者窗口创建成功：{__producer_name}")
                break

    def start_consumer(self):
        """
        启动消费者
        @return:
        """
        while True:
            __consumer_name = self.generate_random_name()
            if __consumer_name not in list(self.consumer_windows.keys()):
                consumer_window = WindowConsumer(num_consumer=__consumer_name, parent=self)

                self.consumer_windows[__consumer_name] = consumer_window
                consumer_window.show()

                consumer_window.installEventFilter(self)
                consumer_window.start_snap_timer()

                global_log.debug(f"消费者窗口创建成功：{__consumer_name}")
                break

        windows = self.consumer_windows.values()
        for i in windows:
            global_log.debug(i.maximumSize())

    def closeEvent(self, event) -> None:
        """
        关闭窗口
        @param event:
        @return:
        """
        if len(self.producer_windows) > 0 or len(self.consumer_windows) > 0:
            __quit = QtWidgets.QMessageBox.question(
                self,
                "消息",
                f"创建了{len(self.producer_windows)}个生产者，{len(self.consumer_windows)}个消费者，是否全部关闭?\n"
                "确认退出？",
                QtWidgets.QMessageBox.Yes | QtWidgets.QMessageBox.No,
                QtWidgets.QMessageBox.No,
            )
            if __quit == QtWidgets.QMessageBox.Yes:
                producer_windows = list(self.producer_windows.keys())
                consumer_windows = list(self.consumer_windows.keys())
                for producer in producer_windows:
                    self.producer_windows[producer].close()
                    event.accept()
                    global_log.debug(f"关闭生产者窗口：{producer}")
                    time.sleep(0.1)
                for consumer in consumer_windows:
                    self.consumer_windows[consumer].close()
                    event.accept()
                    global_log.debug(f"关闭消费者窗口：{consumer}")
                    time.sleep(0.1)
                event.accept()
                global_log.debug("关闭主窗口")
        else:
            # 弹窗确认是否关闭
            __quit = QtWidgets.QMessageBox.question(
                self,
                "消息",
                "确认退出？",
                QtWidgets.QMessageBox.Yes | QtWidgets.QMessageBox.No,
                QtWidgets.QMessageBox.No,
            )
            if __quit == QtWidgets.QMessageBox.Yes:
                event.accept()
                global_log.debug("关闭主窗口")

    def remove_producer_window(self, producer_window):
        """"""
        self.producer_windows.pop(producer_window)
        global_log.info(f"生产者{producer_window}已关闭。剩余生产者: {self.producer_windows.keys()}")

    def remove_consumer_window(self, consumer_window):
        """"""
        self.consumer_windows.pop(consumer_window)
        global_log.info(f"消费者{consumer_window}已关闭。剩余消费者: {self.consumer_windows.keys()}")

    def eventFilter(self, source, event):
        global_log.debug(f"source: {source}, event: {event.type()}")
        self.consumer_windows_obj = list(self.consumer_windows.values())
        self.producer_windows_obj = list(self.producer_windows.values())
        if source in self.consumer_windows_obj and event.type() == event.Move:
            source.snap_to_edge()
            return super().eventFilter(source, event)
        if source in self.producer_windows_obj and event.type() == event.Move:
            source.snap_to_edge()
            return super().eventFilter(source, event)

    @staticmethod
    def generate_random_name(length: int = 8):
        """
        生成随机名称 返回大写字母
        @param length:
        @return:
        """
        return ''.join(random.choices(string.ascii_uppercase, k=length)).upper()


if __name__ == "__main__":
    app = QApplication(sys.argv)
    window = MainWindow()
    window.show()
    sys.exit(app.exec_())
